#include "SocketPollDemo.h"
#include <iostream>

using namespace std;

#if __linux__ | __APPLE__ 

#include <poll.h>
#include "ThreadPool.h"

using namespace common;

SOCKET createPollService(int port)
{
    // 1. 创建 Socket
    SOCKET serverSocket = socket(AF_INET, SOCK_STREAM, 0);
    if (serverSocket == -1)
    {
        cerr << "创建失败！！" << endl;
        return -1;
    }

    // 2. bind IP端口
    sockaddr_in serverAddr;
    serverAddr.sin_family = AF_INET;
    serverAddr.sin_addr.s_addr = INADDR_ANY;
    serverAddr.sin_port = htons(port);
    int bindRet = ::bind(serverSocket, (sockaddr *)&serverAddr, sizeof(serverAddr));
    if (bindRet == -1)
    {
        cerr << "绑定端口失败！！！" << endl;
        compatCloseSocket(serverSocket);
        return -1;
    }

    // 3. listen
    int listenRet = listen(serverSocket, 1024);
    if (listenRet == -1)
    {
        cerr << "监听失败！！！" << endl;
        compatCloseSocket(serverSocket);
        return -1;
    }

    return serverSocket;
}

void simplePollDemo()
{
    SOCKET serverSocket = createPollService(9999);

    // 4. 使用 poll 处理 接受链接、读取数据、写入数据等操作
    pollfd fds[1024];
    fds[0].fd = serverSocket;
    fds[0].events = POLLIN;
    
    for (int i = 1; i < 1024; ++i)
    {
        fds[i].fd = -1;
        fds[i].events = POLLIN;
    }

    int maxIndex = 0;

    while (true)
    {
        int count = poll(fds, maxIndex + 1, -1);
        cout << "已有" << count << "就绪"<< endl;
        if (count == -1)
        {
            cerr << "poll 异常" << endl;
            break;
        }

        if (fds[0].revents & POLLIN)
        {
            sockaddr_in clientAddr;
            addrlen clientAddrLength = sizeof(clientAddr);
            SOCKET clientSocket = accept(fds[0].fd, (sockaddr *)&clientAddr, &clientAddrLength);
            if (clientSocket == -1)
            {
                cerr << "接入客户端链接异常！！！" << endl;
                break;
            }

            cout << "客户端IP：" << inet_ntoa(clientAddr.sin_addr) << "；端口：" << ntohs(clientAddr.sin_port) << "已连接" << endl;

            for (int i = 1; i < 1024; ++i)
            {
                if (fds[i].fd == -1)
                {
                    fds[i].fd = clientSocket;
                    maxIndex = i > maxIndex ? i : maxIndex;
                    break;
                }
            }
        }

        for (int i = 1; i <= maxIndex; ++i)
        {
            if (fds[i].revents & POLLIN)
            {
                char readBuff[1024];
                int len = recv(fds[i].fd, readBuff, sizeof(readBuff), 0);
                if (len == -1)
                {
                    // 读取失败
                    cerr << "读取失败异常: " << endl;
                    continue;
                }
                else if (len == 0)
                {
                    // 断开链接
                    cout << "服务器断开链接！" << endl;
                    compatCloseSocket(fds[i].fd);
                    fds[i].fd = -1;
                    continue;
                }
                else
                {
                    // 读取成功
                    cout << "服务器接受到数据: " << readBuff << endl;
                    // 回复给客户端
                    for (int i = 0; i < len; ++i)
                    {
                        readBuff[i] = toupper(readBuff[i]);
                    }
                    int sendRet = send(fds[i].fd, readBuff, strlen(readBuff) + 1, 0);
                    if (sendRet == -1)
                    {
                        // 读取失败
                        cerr << "回复失败: " << endl;
                    }
                }
            }
        }
    }

    for (int i = 0; i <= maxIndex; ++i)
    {
        if (fds[i].fd != -1)
        {
            cout << "关闭 Socket 链接：" << i << endl;
            compatCloseSocket(fds[i].fd);
        }
    }
}

void threadPollDemo()
{
    SOCKET serverSocket = createPollService(9999);
    
    // 4. 使用 poll 处理 接受链接、读取数据、写入数据等操作
    pollfd fds[1024];
    fds[0].fd = serverSocket;
    fds[0].events = POLLIN;
    
    for (int i = 1; i < 1024; ++i)
    {
        fds[i].fd = -1;
        fds[i].events = POLLIN;
    }

    int maxIndex = 0;

    ThreadPool threadPool(3, 10);
    static mutex pollMutex;
    struct PollInfo {
        SOCKET fd;
        pollfd* fds[1024];
        int* maxIndex;
    };

    TaskCallBack readTask = [](void *const arg) -> void {
        PollInfo* pollInfo = (PollInfo*)arg;

        char readBuff[1024];
        int len = recv(pollInfo->fd, readBuff, sizeof(readBuff), 0);
        if (len == -1)
        {
            // 读取失败
            cerr << "读取失败异常: " << endl;
            free(pollInfo);
            return;
        }
        else if (len == 0)
        {
            // 断开链接
            cout << "服务器断开链接！" << endl;
            compatCloseSocket(pollInfo->fd);
            free(pollInfo);
            return;
        }

        pollMutex.lock();
        for (int i = 1; i < 1024; ++i)
        {
            if ((*pollInfo->fds)[i].fd == -1)
            {
                (*pollInfo->fds)[i].fd = pollInfo->fd;
                *pollInfo->maxIndex = i > *pollInfo->maxIndex ? i : *pollInfo->maxIndex;
                break;
            }
        }
        pollMutex.unlock();

        // 读取成功
        cout << "服务器接受到数据: " << readBuff << endl;
        // 回复给客户端
        for (int i = 0; i < len; ++i)
        {
            readBuff[i] = toupper(readBuff[i]);
        }
        int sendRet = send(pollInfo->fd, readBuff, strlen(readBuff) + 1, 0);
        if (sendRet == -1)
        {
            // 读取失败
            cerr << "回复失败: " << endl;
        }
        free(pollInfo);
    };

    while (true)
    {
        pollMutex.lock();
        int count = poll(fds, maxIndex + 1, 0);
        pollMutex.unlock();

        if (count == -1)
        {
            cerr << "poll 异常" << endl;
            break;
        } else if (count == 0) {
            continue;
        }

        if (fds[0].revents & POLLIN)
        {
            sockaddr_in clientAddr;
            addrlen clientAddrLength = sizeof(clientAddr);
            SOCKET clientSocket = accept(fds[0].fd, (sockaddr *)&clientAddr, &clientAddrLength);
            if (clientSocket == -1)
            {
                cerr << "接入客户端链接异常！！！" << endl;
                break;
            }

            cout << "客户端IP：" << inet_ntoa(clientAddr.sin_addr) << "；端口：" << ntohs(clientAddr.sin_port) << "已连接" << endl;

            pollMutex.lock();
            for (int i = 1; i < 1024; ++i)
            {
                if (fds[i].fd == -1)
                {
                    fds[i].fd = clientSocket;
                    maxIndex = i > maxIndex ? i : maxIndex;
                    break;
                }
            }
            pollMutex.unlock();
        }

        pollMutex.lock();
        for (int i = 1; i <= maxIndex; ++i)
        {
            if (fds[i].revents & POLLIN)
            {
                PollInfo* pollInfo = (PollInfo*) malloc(sizeof(PollInfo));
                pollInfo->fd = fds[i].fd;
                *pollInfo->fds = fds;
                pollInfo->maxIndex = &maxIndex;
                fds[i].fd = -1;
                threadPool.addTask(readTask, pollInfo);
            }
        }
        pollMutex.unlock();
    }

    // 在关闭所有链接前需要先关闭正在运行中的线程否则可能会存在问题，这里暂不处理了。

    for (int i = 0; i <= maxIndex; ++i)
    {
        if (fds[i].fd != -1)
        {
            cout << "关闭 Socket 链接：" << i << endl;
            compatCloseSocket(fds[i].fd);
        }
    }
}

#endif

void socketPollDemo()
{
    // 对于 socket 并发编程存在多个阻塞点，这些组赛点相互互斥，导致需要使用多线程来解决，当并发两居高时会导致大量线程被开辟从而造成内存和系统资源开销影响服务器负荷。
    // 但其实中间有诸多客户端链接可能只是保持链接并没有过于频繁的操作，这样的线程开辟造成了资源的浪费，如果处理线程可以得到复用则会大大奖励这种负荷开销实现高性能并发。
    // 在 C++ socket 编程中提供了 select、poll、epoll 三种方式，其中，select、poll 内部使用链表维护，epoll 使用红黑树维护，所以：
    //  1. 在性能上 select、poll 要略逊于 epoll；
    //  2. 在可检测连接数上 select 也有限制，最大时 1024，超过 1024 就不行了（这个取决于系统配置），而 epoll、poll 没有这方面限制可以无限大；
    //  3. select 支持 windows 和 mac、linux；但 poll 和 epoll 并不支持 windows。
    // 注意：这里的可检测连接数并不是并发连接数，可以并发链接的数量取决于ip和端口组合数。这里是指在链接到服务器上的链接里去检测那些有传输数据那些没有。
    //      也就是说可以间接2000个客户端，但是超过1024后面的可能得不到检测。
    // 总结：所以我们需要着重理解 select 和 epoll 因为在 linux 下 epoll 有更好得优势，而考虑跨平台性时只能选择 select。

    // poll 函数
    // poll 的机制与 select 类似，与 select 在本质上没有多大差别，使用方法也类似，下面的是对于二者的对比：
    //  * 内核对应文件描述符的检测也是以线性的方式进行轮询，根据描述符的状态进行处理;
    //  * poll 和 select 检测的文件描述符集合会在检测过程中频繁的进行用户区和内核区的拷贝，它的开销随着文件描述符数量的增加而线性增大，从而效率也会越来越低;
    //  * select 检测的文件描述符个数上限是 1024，poll 没有最大文件描述符数量的限制；
    //  * select 可以跨平台使用，poll 只能在 Linux 平台使用
    //
    // poll函数的函数原型如下：
    //    #include <poll.h>
    //    // 每个委托poll检测的fd都对应这样一个结构体
    //    struct pollfd {
    //        int   fd;         /* 委托内核检测的文件描述符 */
    //        short events;     /* 委托内核检测文件描述符的什么事件 */
    //        short revents;    /* 文件描述符实际发生的事件 -> 传出 */
    //    };
    //    
    //    struct pollfd myfd[100];
    //    int poll(struct pollfd *fds, nfds_t nfds, int timeout);
    //
    // 函数参数：
    // * struct pollfd *fds：这是一个 struct pollfd 类型的数组, 里边存储了待检测的文件描述符的信息，这个数组中有三个成员：
    //   - fd：委托内核检测的文件描述符
    //   - events：委托内核检测的fd事件（输入、输出、错误），每一个事件有多个取值
    //   - revents：这是一个传出参数，数据由内核写入，存储内核检测之后的结果
    //   ┌───────┬────────────┬────────┬─────────┬─────────────────────────────────┐
    //   │ event │    value   │ events │ revents │               说明               
    //   ├───────┼────────────┼────────┼─────────┼─────────────────────────────────┤
    //   │       │ POLLIN     │   √    │    √    │       普通或优先带数据可读
    //   │ read  │ POLLRDNORM │   √    │    √    │           普通数据可读            
    //   │       │ POLLRDBAND │   √    │    √    │         优先级带数据可读          
    //   │       │ POLLPRI    │   √    │    √    │         高优先级数据可读          
    //   ├───────┼────────────┼────────┼─────────┼─────────────────────────────────┤
    //   │       │ POLLOUT    │   √    │    √    │       普通或优先带数据可写        
    //   │ write │ POLLWRNORM │   √    │    √    │           普通数据可写            
    //   │       │ POLLWRBAND │   √    │    √    │         优先级带数据可写          
    //   ├───────┼────────────┼────────┼─────────┼─────────────────────────────────┤
    //   │       │ POLLERR    │        │    √    │             发生错误              
    //   │ error │ POLLHUP    │        │    √    │             发生挂起              
    //   │       │ POLLNVAL   │        │    √    │        描述不是打开的文件          
    //   └───────┴────────────┴────────┴─────────┴─────────────────────────────────┘
    // * nfds_t nfds：这是第一个参数数组中最后一个有效元素的下标 + 1（也可以指定第一个参数所引用数组的元素总个数）
    // * int timeout：指定 poll 函数的阻塞时长:
    //   - -1：一直阻塞，直到检测的集合中有就绪的文件描述符（有事件产生）解除阻塞
    //   - 0：不阻塞，不管检测集合中有没有已就绪的文件描述符，函数马上返回
    //   - 大于0：阻塞指定的毫秒（ms）数之后，解除阻塞
    //
    // 函数返回值：
    //  * 失败： 返回-1
    //  * 成功：返回一个大于0的整数，表示检测的集合中已就绪的文件描述符的总个数
    //
    // 使用 poll 和 select 进行 IO 多路转接的处理思路是完全相同的，但是使用 poll 编写的代码看起来会更直观一些，
    // select 使用的位图的方式来标记要委托内核检测的文件描述符（每个比特位对应一个唯一的文件描述符），
    // 并且对这个 fd_set 类型的位图变量进行读写还需要借助一系列的宏函数，操作比较麻烦。
    // 而 poll 直接将要检测的文件描述符的相关信息封装到了一个结构体 struct pollfd 中，我们可以直接读写这个结构体变量。
    // 另外 poll 的第二个参数有两种赋值方式，但是都和第一个参数的数组有关系：
    //  * 使用第一个参数所引用数组的元素个数
    //  * 使用第一个参数所引用数组中存储的最后一个有效元素对应的下标值 + 1
    // 内核会根据第二个参数传递的值对第一个参数所引用数组中的文件描述符进行线性遍历，这一点和 select 也是类似的。


    cout << "案例六：IO多路转接（复用）之poll方案演示开始" << endl;

#ifdef _WIN32
    cout << "windows操作系统平台不支持 poll 模型！！可以查看 select 模型以及 IOCP 模型" << endl;
#else
    cout << "请选择演示哪一种实现：" << endl;
    cout << "1. poll 模型的简单单线程实现多路转接并发处理" << endl;
    cout << "2. poll 模型的基于线程池复用多路转接并发处理" << endl;

    char commond;
    cin >> commond;
    switch (tolower(commond))
    {
    case '1':
        simplePollDemo();
        break;
    case '2':
        threadPollDemo();
        break;
    }
#endif
}